package com.rosense.nfvo.util;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.BufferedMutator;
import org.apache.hadoop.hbase.client.BufferedMutatorParams;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

import com.rosense.commons.io.IOUtils;


public class Kbs2HbaseUtil {
	
	public static void main(String[] args) {
	
			int start=1000;
			int partition=200;
			
			 List<String> list=new ArrayList<>();
	 /*
			 list.add("user_5gznyy:tdr_ims_inf_register_sip");
			 list.add("user_5gznyy:tdr_ims_register_sip");
			 list.add("user_5gznyy:cdr_ims_inf_call_sip");
			 list.add("user_5gznyy:cdr_ims_mo_call_leg_sip");
			 list.add("user_5gznyy:cdr_ims_mt_call_leg_sip");
			 list.add("user_5gznyy:cdr_ims_call_delay");
			 list.add("user_5gznyy:cdr_volte_voice_quality");
		//	 list.add("user_5gznyy:tdr_ims_dns");
			 list.add("user_5gznyy:tdr_ims_enum");
		//	 list.add("user_5gznyy:tdr_ims_inf_ip_message_sip");
		//	 list.add("user_5gznyy:tdr_ims_ip_message_sip");
		//	 list.add("user_5gznyy:tdr_ims_inf_ss_sip");
		//	 list.add("user_5gznyy:tdr_ims_ss_sip");
		//	 list.add("user_5gznyy:tdr_ims_inf_miscellaneous_sip");
		//	 list.add("user_5gznyy:tdr_ims_miscellaneous_sip");
			 list.add("user_5gznyy:tdr_ims_hss_trans");
			 list.add("user_5gznyy:tdr_ims_bad_ipmos_info");
		*/
			 list.add("user_5gznyy:tdr_ims_test_v2");
			 for (int i = 0; i < list.size(); i++) {
					Kbs2HbaseUtil.createTable(list.get(i),new String[]{Constants.HBASE_CALL}, Kbs2HbaseUtil.createRegions(start, partition));

			}
			
		
		
	}
	
	public static Connection getConnection(){
		try {
			Configuration conf = HBaseConfiguration.create();
			conf.addResource(new Path(Constants.hbase_conf));
			conf.addResource(new Path(Constants.core_conf));
			UserGroupInformation.setConfiguration(conf);
			UserGroupInformation.loginUserFromKeytab(Constants.user,Constants.keytab);

			Connection connection = ConnectionFactory.createConnection(conf);
			System.out.println("get conn success");
			return connection;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	/**
     * 閼惧嘲褰噐egion 娴滃瞼娣弫鎵矋 -
     *
     * @param start     - 鐠у嘲顫怚D
     * @param partition - 閸掑棗灏弫浼村櫤
     * @return
     */
    public static byte[][] createRegions(int start, int partition) {
        byte[][] splitKeys = new byte[partition][];
        for (int i = 0; i < partition; i++) {
            splitKeys[i] = Bytes.toBytes((i + start) + "");
        }
        return splitKeys;
    }

    public static void createTable(String tableName, String[] family, byte[][] regions) {
        System.out.println("start create table ......" + tableName);
        Connection conn = null;
        try {
            // 娴犲氦绻涢幒銉よ厬閺嬪嫰锟界姳绔存稉鐙L閹垮秳缍旈崳锟�
            conn = getConnection();
            System.out.println("connection is " + conn);
            Admin admin = conn.getAdmin();
            System.out.println("admin is " + admin);
            TableName tName = TableName.valueOf(tableName);
//            if (!admin.tableExists(tName)) {
                System.out.println(tableName + " is not exist,Creat....");
                // 閸掓稑缂撴稉锟芥稉顏囥�冪�规矮绠熼幓蹇氬牚鐎电钖�
                HTableDescriptor tableDescriptor = new HTableDescriptor(tName);
                tableDescriptor.setRegionSplitPolicyClassName("org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy");
		//		tableDescriptor.setMaxFileSize(20*1024*1024*1024);
//                tableDescriptor.setMaxFileSize(1024 * 1024 * 1024);

                for (String s : family) {
                    // 閸掓稑缂撻崚妤佹鐎规矮绠熼幓蹇氬牚鐎电钖�
                    HColumnDescriptor hColumnDescriptor_1 = new HColumnDescriptor(s);
                    // 閸掓稑缂撻崚妤佹閻楀牊婀伴張锟芥径褎鏆�
//				hColumnDescriptor_1.setMaxVersions(Integer.MAX_VALUE);
                    hColumnDescriptor_1.setCompressionType(Algorithm.GZ);
                    hColumnDescriptor_1.setTimeToLive(7 * 24 * 60 * 60);
                    // 鐏忓棗鍨弮蹇撶暰娑斿淇婇幁顖氼嚠鐠炩剝鏂侀崗銉ㄣ�冪�规矮绠熺�电钖勬稉锟�
                    tableDescriptor.addFamily(hColumnDescriptor_1);
                }

                // 閻⑩暊dl閹垮秳缍旈崳銊ヮ嚠鐠炩槄绱癮dmin 閺夈儱缂撶悰锟�
                admin.createTable(tableDescriptor, regions);
                // 閸忔娊妫存潻鐐村复
                admin.close();
//            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeIO(conn);
        }
        System.out.println("end create table ......");
    }
    
    public static void insertData(List<Put> lp, String tableName) throws IOException {
        Connection conn = null;
        try {
            long time = System.currentTimeMillis();
            int total = lp.size();
            System.out.println("start insert " + total + "閸涚厧褰╠ata ......");
            if (total == 0) {
                return;
            }

            conn = getConnection();
            TableName tname = TableName.valueOf(tableName);
            BufferedMutatorParams params = new BufferedMutatorParams(tname).writeBufferSize(50 * 1024 * 1024);
            BufferedMutator mutator = conn.getBufferedMutator(params);
            mutator.mutate(lp);
            mutator.flush();
            mutator.close();
            System.out.println("end insert data ......閼版妞�" + (System.currentTimeMillis() - time) / 1000 + "缁夛拷");

        } catch (Exception e) {
            System.out.println(e.getMessage());
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
                conn = null;
            }
        }
    }
    
    
	//查询单列族多列的
	
	public static List<Map<String,Object> > selectStartToEndFilter(String tablename,String startrow,String endrow) {
		
		List<Map<String,Object> > dataL = new ArrayList<>();
		 Connection conn = null;
		try{
			
		 String family="info";
	//	 String column="value";
		  conn = getConnection();
		  System.out.println("conn****"+conn);
		  Table table = conn.getTable(TableName.valueOf(tablename));
		  System.out.println("table****"+table);
	    
	   	 Scan scan = new Scan();
	   	 scan.setStartRow(startrow.getBytes());
	   	 scan.setStopRow(endrow.getBytes());
	   	 ResultScanner scanner = table.getScanner(scan);
	 	
		 for(Result result:scanner){
			 Map<String,Object> data = new HashMap<String,Object>();
			
			// 获取该列族的所有的列限定符，存放在 cols 中
             Map<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes(family));
             System.out.println("familyMap****"+familyMap.size());
             ArrayList<String> cols = new ArrayList<String>();
             for(Map.Entry<byte[], byte[]> entry:familyMap.entrySet()){
                 String key = Bytes.toString(entry.getKey());
                 String value = new String(result.getValue(family.getBytes(), key.getBytes()));
               
                 data.put(key.toUpperCase(), value);
             }
			 dataL.add(data);
	       }	
		}catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}
		 return dataL;  	
	   }
	
    
	
//	public static void createTable_v2(String tableName, String family, byte[][] regions) {
//		System.out.println("start create table ......");
//		 Connection conn = null;
//		try {
//			// 从连接中构造一个DDL操作器
//			 conn = getConnection();
//			
//			Admin admin = conn.getAdmin();
//			
//			TableName tName = TableName.valueOf(tableName);
//			if (!admin.tableExists(tName)) {
//				System.out.println(tableName + " is not exist,Creat....");
//				// 创建一个表定义描述对象
//				HTableDescriptor tableDescriptor = new HTableDescriptor(tName);
//				tableDescriptor.setRegionSplitPolicyClassName(
//						"org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy");
//			     tableDescriptor.setMaxFileSize(30 * 1024 * 1024 * 1024);
//
//				// 创建列族定义描述对象
//				HColumnDescriptor hColumnDescriptor_1 = new HColumnDescriptor(family);
//				// 创建列族版本最大数
//				// hColumnDescriptor_1.setMaxVersions(Integer.MAX_VALUE);
//				hColumnDescriptor_1.setCompressionType(Algorithm.GZ);
//				//设置保存时间
//				hColumnDescriptor_1.setTimeToLive(7*24*60*60);
//				
//				// 将列族定义信息对象放入表定义对象中
//				tableDescriptor.addFamily(hColumnDescriptor_1);
//
//				// 用ddl操作器对象：admin 来建表
//				admin.createTable(tableDescriptor, regions);
//				// 关闭连接
//				admin.close();
//			}
//
//			conn.close();
//			conn = null;
//		} catch (MasterNotRunningException e) {
//			e.printStackTrace();
//		} catch (ZooKeeperConnectionException e) {
//			e.printStackTrace();
//		} catch (IOException e) {
//			e.printStackTrace();
//		}
//		System.out.println("end create table ......");
//	}
	
	
	
	  
	
	
}
